<?php

declare(strict_types=1);

namespace JfbQueue;

use JfbQueue\Driver\DriverInterface;
use Hyperf\Context\ApplicationContext;
use Hyperf\Contract\ConfigInterface;
use Hyperf\Process\Annotation\Process;
use Hyperf\Process\ProcessManager;
use Hyperf\Redis\RedisFactory;
use Hyperfx\Framework\Exception\AppException;
use Hyperfx\Framework\Logger\Logx;
use JfbQueue\Message\MessageInterface;
use Hyperfx\Utils\AppEnvUtil;
use Hyperfx\Utils\RunModeUtil;
use Psr\Container\ContainerInterface;
use Hyperf\Process\AbstractProcess;


abstract class AbstractConsumer extends AbstractProcess {
    
    public string $queue = 'default';
    protected array $config;
    
    protected int $restartInterval = 1;

    private array $streams = [];

    private string $consumer;

    private DriverInterface $drvier;

    private int $maxAttempts = 5;

    private bool $isEnabledReload = false;

    /* @param array $payloads  */
    abstract protected function consume(array $payloads): array;

    /* @param array $payloads  */
    abstract protected function fail(array $payloads): void;

    public function __construct(ContainerInterface $container)
    {
        parent::__construct($container);

        $driver = ApplicationContext::getContainer()->get(DriverFactory::class);
        $this->isEnabledReload = $driver->isEnabledReload();
        if ($this->isEnabledReload) {
            $this->config = $driver->getConfigByQueue($this->queue);
            $this->drvier = $driver->get($this->queue);
            $this->maxAttempts = $this->drvier->getMaxAttempts();

            $this->name = "async_queue_x.{$this->queue}";
            $this->nums = $this->config['processes'] ?? 1;
        }
    }

    public function isEnable($server): bool {
        return $this->isEnabledReload;
    }

    private function pop() {
        try {
            $messages = $this->drvier->pop();
            if (false === $messages) {
                return [[], []];
            }

            $arrConsumeMessages = [];
            $arrFailMessages = [];
            /* @var $message MessageInterface */
            foreach ($messages as $message) {
                if ($message->getAttempts() > $this->maxAttempts) {
                    Logx::get()->debug('消息处理失败达到最大次数', [
                        'message_body' => $message->getBody(),
                    ]);
                    $arrFailMessages[] = $message;
                } else {
                    $arrConsumeMessages[] = $message;
                }
            }
            return [$arrConsumeMessages, $arrFailMessages];

        } catch (\Throwable $e) {
            Logx::get()->alert('POP消息失败', [
                'error_message' => $e->getMessage(),
                'line' => $e->getLine(),
                'code' => $e->getCode(),
                'trace' => $e->getTraceAsString()
            ]);
        }
        return [[], []];
    }

    public function handle(): void
    {
        $messageCount = 0;
        $maxMessages = $this->config['max_messages'];
        $this->drvier->init();

        while (ProcessManager::isRunning()) {
            $traceId = Logx::nextTraceId();
            if ($maxMessages > 0 && $messageCount >= $maxMessages) {
                break;
            }

            // pop messages
            [$arrConsumeMessages, $arrFailMessages] = $this->pop();

            // consume messages
            $this->consumeMessaages($arrConsumeMessages, $traceId);
            
            // consume fail messages
            $this->consumeFailMessaages($arrFailMessages, $traceId);
            
            ++$messageCount;
        }
    }

    private function consumeMessaages(array $messages, $traceId) {
        if (empty($messages)) {
            return;
        }
        try {
            $callback = parallel([function() use ($traceId, $messages) {
                Logx::nextTraceId($traceId);
                // 正常消息
                return $this->consume($messages);
            }]);

            $status = $callback[0][0] ?? '';

            switch ($status) {
                // 成功
                case Result::ACK: {
                    $this->drvier->ack($messages);
                    break;
                }
                // 失败
                case Result::FAIL: {
                    $this->drvier->fail($messages);
                    Logx::get()->error('处理消息失败', [
                        'message' => var_export($messages, true),
                    ]);
                    break;
                }
                // 重试
                case Result::REQUEUE: {
                    $retryTime = (int) $callback[0][1] ?? 5;
                    $this->drvier->retry($messages, $retryTime);
                    break;
                }
                default:{
                    Logx::get()->alert('返回状态错误', [
                        'message' => var_export($messages, true),
                    ]);
                }
            }

        } catch (\Throwable $e) {
            $this->drvier->fail($messages);
            Logx::get()->alert('消息处理失败', [
                'messages' => var_export($messages, true),
                'error_message' => $e->getMessage(),
                'line' => $e->getLine(),
                'code' => $e->getCode(),
                'trace' => $e->getTraceAsString()
            ]);
        }
    }

    /**
     * 消费失败的消息
     */
    private function consumeFailMessaages(array $messages, $traceId) {
        if (empty($messages)) {
            return;
        }
        try {
            parallel([function() use ($traceId, $messages) {
                Logx::nextTraceId($traceId);
                // 正常消息
                $this->fail($messages);
                $this->drvier->ack($messages);
            }]);

        } catch (\Throwable $e) {
            $this->drvier->fail($messages);
            Logx::get()->alert('消息失效又报错了', [
                'messages' => var_export($messages, true),
                'error_message' => $e->getMessage(),
                'line' => $e->getLine(),
                'code' => $e->getCode(),
                'trace' => $e->getTraceAsString()
            ]);
        }
    }
}